package io.kiki.sba.registry.server.metadata.bootstrap;


import io.kiki.sba.registry.server.metadata.bootstrap.config.MultiClusterMetaServerConfig;
import io.kiki.sba.registry.server.metadata.bootstrap.config.ServerConfig;
import io.kiki.sba.registry.task.KeyedThreadPoolExecutor;
import io.kiki.sba.registry.task.MetricsableThreadPoolExecutor;
import io.kiki.sba.registry.util.NamedThreadFactory;
import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;


public class ExecutorManager {

    private static final Logger logger = LoggerFactory.getLogger(ExecutorManager.class);
    private static final String MULTI_CLUSTER_CONFIG_RELOAD_EXECUTOR = "MULTI_CLUSTER_CONFIG_RELOAD_EXECUTOR";
    private static final String REMOTE_SLOT_SYNCER_EXECUTOR = "REMOTE_SLOT_SYNCER_EXECUTOR";
    private static final String REMOTE_CLUSTER_HANDLER_EXECUTOR = "REMOTE_CLUSTER_HANDLER_EXECUTOR";
    private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;
    @Getter
    private final ThreadPoolExecutor multiClusterConfigReloadExecutor;
    @Getter
    private final ThreadPoolExecutor remoteClusterHandlerExecutor;
    @Getter
    private final KeyedThreadPoolExecutor remoteSlotSyncerExecutor;
    private Map<String, ThreadPoolExecutor> reportExecutors = new HashMap<>();

    private Map<String, KeyedThreadPoolExecutor> keyExecutors = new HashMap<>();

    public ExecutorManager(ServerConfig serverConfig, MultiClusterMetaServerConfig multiClusterMetaServerConfig) {
        scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(serverConfig.getMetaSchedulerPoolSize(), new NamedThreadFactory("MetaScheduler"));

        multiClusterConfigReloadExecutor = reportExecutors.computeIfAbsent(MULTI_CLUSTER_CONFIG_RELOAD_EXECUTOR, k -> new MetricsableThreadPoolExecutor(MULTI_CLUSTER_CONFIG_RELOAD_EXECUTOR, multiClusterMetaServerConfig.getMultiClusterConfigReloadWorkerSize(), multiClusterMetaServerConfig.getMultiClusterConfigReloadWorkerSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(multiClusterMetaServerConfig.getMultiClusterConfigReloadMaxBufferSize()), new NamedThreadFactory(MULTI_CLUSTER_CONFIG_RELOAD_EXECUTOR, true), (r, executor) -> {
            String msg = String.format("Multi cluster reload task(%s) %s rejected from %s, just ignore it.", r.getClass(), r, executor);
            logger.error(msg);
        }));

        remoteSlotSyncerExecutor = keyExecutors.computeIfAbsent(REMOTE_SLOT_SYNCER_EXECUTOR, k -> new KeyedThreadPoolExecutor(REMOTE_SLOT_SYNCER_EXECUTOR, multiClusterMetaServerConfig.getRemoteSlotSyncerExecutorPoolSize(), multiClusterMetaServerConfig.getRemoteSlotSyncerExecutorQueueSize()));

        remoteClusterHandlerExecutor = reportExecutors.computeIfAbsent(REMOTE_CLUSTER_HANDLER_EXECUTOR, k -> new MetricsableThreadPoolExecutor(REMOTE_CLUSTER_HANDLER_EXECUTOR, multiClusterMetaServerConfig.getRemoteClusterHandlerCoreSize(), multiClusterMetaServerConfig.getRemoteClusterHandlerMaxSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(multiClusterMetaServerConfig.getRemoteClusterHandlerMaxBufferSize()), new NamedThreadFactory(REMOTE_CLUSTER_HANDLER_EXECUTOR, true)));
    }

    public void startScheduler() {
    }

    public void stopScheduler() {
        if (scheduledThreadPoolExecutor != null && !scheduledThreadPoolExecutor.isShutdown()) {
            scheduledThreadPoolExecutor.shutdown();
        }
    }


}
